package drds.server.net;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Strings;

import drds.server.util.CurrentTimeMillis;

/**
 * 网络处理
 * 
 * @author 曹正辉<330937205@qq.com>
 * 
 */
public abstract class SessionContext$ReadWrite implements ISessionContext {
	protected static final Logger LOGGER = LoggerFactory.getLogger(SessionContext$ReadWrite.class);
	protected final AtomicBoolean isClosed;
	/** 网络信息 */
	protected String host;
	protected final SocketChannel socketChannel;
	protected long id;
	/** 各种处理器 */
	private final IReadWrite readWrite;
	protected SessionContextGroupShareResources processor;
	protected Handler handler;

	protected int packetHeaderSize;
	protected int maxPacketSize;
	protected volatile ByteBuffer readByteBuffer;
	protected volatile ByteBuffer writeByteBuffer;

	protected final ConcurrentLinkedQueue<ByteBuffer> writeByteBufferQueue = new ConcurrentLinkedQueue<ByteBuffer>();

	protected volatile int readByteBufferOffset;
	protected long lastLargeMessageTime;

	/** 有限性测试数据 */
	protected long startTime;
	protected long lastReadTime;
	protected long lastWriteTime;
	private long idleTimeout;
	protected int writeAttempts;
	/** 网络流量 */
	protected long netInBytesSize;
	protected long netOutBytesSize;

	/** 压缩和解压缩 */
	protected final ConcurrentLinkedQueue<byte[]> decompressUnfinishedDataQueue = new ConcurrentLinkedQueue<byte[]>();
	protected final ConcurrentLinkedQueue<byte[]> compressUnfinishedDataQueue = new ConcurrentLinkedQueue<byte[]>();

	public SessionContext$ReadWrite(SocketChannel socketChannel) {
		this.socketChannel = socketChannel;
		this.isClosed = new AtomicBoolean(false);
		//
		this.readWrite = new ReadWrite(this);
		//
		this.startTime = CurrentTimeMillis.currentTimeMillis();
		this.lastReadTime = startTime;
		this.lastWriteTime = startTime;
	}

	public long getIdleTimeout() {
		return this.idleTimeout;
	}

	public IReadWrite getReadWrite() {
		return this.readWrite;
	}

	public void setIdleTimeout(long idleTimeout) {
		this.idleTimeout = idleTimeout;
	}

	public String getHost() {
		return this.host;
	}

	public void setHost(String host) {
		this.host = host;
	}

	public long getId() {
		return this.id;
	}

	public void setId(long id) {
		this.id = id;
	}

	public boolean isIdleTimeout() {
		return CurrentTimeMillis.currentTimeMillis() > Math.max(lastWriteTime, lastReadTime) + idleTimeout;
	}

	public SocketChannel getSocketChannel() {
		return this.socketChannel;
	}

	public int getPacketHeaderSize() {
		return this.packetHeaderSize;
	}

	public void setPacketHeaderSize(int packetHeaderSize) {
		this.packetHeaderSize = packetHeaderSize;
	}

	public int getMaxPacketSize() {
		return this.maxPacketSize;
	}

	public void setMaxPacketSize(int maxPacketSize) {
		this.maxPacketSize = maxPacketSize;
	}

	public long getStartTime() {
		return this.startTime;
	}

	public long getLastReadTime() {
		return this.lastReadTime;
	}

	public void setProcessor(SessionContextGroupShareResources processor) {
		this.processor = processor;
		int size = processor.getByteBufferPool().getBlockSize();
		this.readByteBuffer = processor.getByteBufferPool().allocate(size);
	}

	public long getLastWriteTime() {
		return this.lastWriteTime;
	}

	public long getNetInBytesSize() {
		return this.netInBytesSize;
	}

	public long getNetOutBytesSize() {
		return this.netOutBytesSize;
	}

	public int getWriteAttempts() {
		return this.writeAttempts;
	}

	public SessionContextGroupShareResources getProcessor() {
		return this.processor;
	}

	public ByteBuffer getReadByteBuffer() {
		return this.readByteBuffer;
	}

	public ByteBuffer allocate() {
		int chunkSize = this.processor.getByteBufferPool().getBlockSize();
		ByteBuffer byteBuffer = this.processor.getByteBufferPool().allocate(chunkSize);
		return byteBuffer;
	}

	public final void recycle(ByteBuffer byteBuffer) {
		this.processor.getByteBufferPool().recycle(byteBuffer);
	}

	public void setHandler(Handler handler) {
		this.handler = handler;
	}

	@Override
	public void handle(byte[] bytes) {
		this.handler.handle(bytes);
	}

	public void read() throws IOException {
		this.readWrite.read();
	}

	public void write() throws IOException {
		this.readWrite.write();
	}

	/**
	 * 读取数据并执行相对应的命令
	 */
	public void onReadData(int readDataSize) throws IOException {

		if (this.isClosed.get()) {
			return;
		}
		this.lastReadTime = CurrentTimeMillis.currentTimeMillis();
		if (readDataSize < 0) {
			this.close("io closed");
			return;
		} else if (readDataSize == 0 && !this.socketChannel.isOpen()) {
			this.close("socket closed");
			return;
		}
		this.netInBytesSize += readDataSize;
		this.processor.addNetInBytes(readDataSize);

		// 循环处理字节信息
		int offset = this.readByteBufferOffset;
		int length = 0;
		int position = this.readByteBuffer.position();
		for (;;) {
			length = getPacketLength(this.readByteBuffer, offset);
			if (length == -1) {
				if (offset != 0) {
					this.readByteBuffer = compactReadByteBuffer(this.readByteBuffer, offset);
				} else if (this.readByteBuffer != null && !this.readByteBuffer.hasRemaining()) {
					throw new RuntimeException("invalid readbuffer capacity ,too little buffer size " + readByteBuffer.capacity());
				}
				break;
			}

			if (position >= offset + length && this.readByteBuffer != null) {

				// handle this package
				this.readByteBuffer.position(offset);
				byte[] bytes = new byte[length];
				this.readByteBuffer.get(bytes, 0, length);
				handle(bytes);

				// maybe handle stmt_close
				if (isClosed()) {
					return;
				}

				// offset to next position
				offset += length;

				// reached end
				if (position == offset) {
					// if cur buffer is temper none direct byte buffer and not
					// received large message in recent 30 seconds
					// then change to direct buffer for performance
					if (this.readByteBuffer != null && !this.readByteBuffer.isDirect() && this.lastLargeMessageTime < lastReadTime - 30 * 1000L) { // used
						// temp
						// heap
						recycle(this.readByteBuffer);
						this.readByteBuffer = this.processor.getByteBufferPool().allocate(
						//
								this.processor.getByteBufferPool().getReadByteBufferBlock()//
								);//
					} else {
						if (this.readByteBuffer != null) {
							this.readByteBuffer.clear();
						}
					}
					// no more data ,break
					this.readByteBufferOffset = 0;
					break;
				} else {
					// try next package parse
					this.readByteBufferOffset = offset;
					if (this.readByteBuffer != null) {
						this.readByteBuffer.position(position);
					}
					continue;
				}

			} else {
				// not read whole message package ,so check if buffer enough and
				// compact readbuffer
				if (!this.readByteBuffer.hasRemaining()) {
					this.readByteBuffer = ensureFreeSpaceOfReadBuffer(this.readByteBuffer, offset, length);
				}
				break;
			}
		}
	}

	private ByteBuffer ensureFreeSpaceOfReadBuffer(ByteBuffer byteBuffer, int offset, final int packetLength) {
		// need a large buffer to hold the package
		if (packetLength > this.maxPacketSize) {
			throw new IllegalArgumentException("Packet size over the limit.");
		} else if (byteBuffer.capacity() < packetLength) {

			ByteBuffer newByteBuffer = this.processor.getByteBufferPool().allocate(packetLength);
			this.lastLargeMessageTime = CurrentTimeMillis.currentTimeMillis();
			byteBuffer.position(offset);
			newByteBuffer.put(byteBuffer);
			this.readByteBuffer = newByteBuffer;

			recycle(byteBuffer);
			this.readByteBufferOffset = 0;
			return newByteBuffer;

		} else {
			if (offset != 0) {
				// compact bytebuffer only
				return compactReadByteBuffer(byteBuffer, offset);
			} else {
				throw new RuntimeException(" not enough space");
			}
		}
	}

	private ByteBuffer compactReadByteBuffer(ByteBuffer byteBuffer, int offset) {
		if (byteBuffer == null) {
			return null;
		}
		/**
		 * 压缩数据offset to position
		 */
		byteBuffer.limit(byteBuffer.position());
		byteBuffer.position(offset);
		byteBuffer = byteBuffer.compact();
		this.readByteBufferOffset = 0;
		return byteBuffer;
	}

	public void write(byte[] bytes) {
		ByteBuffer byteBuffer = allocate();
		byteBuffer = writeToByteBuffer(bytes, byteBuffer);
		write(byteBuffer);

	}

	/**
	 * add to queue but not send
	 */
	private final void writeButNotSend(ByteBuffer byteBuffer) {
		this.writeByteBufferQueue.offer(byteBuffer);
	}

	@Override
	public final void write(ByteBuffer byteBuffer) {

		this.writeByteBufferQueue.offer(byteBuffer);

		// if ansyn write finishe event got lock before me ,then writing
		// flag is set false but not start a write request
		// so we check again
		try {
			this.readWrite.write();
		} catch (Exception e) {
			LOGGER.error("write err:", e);
			this.close("write err:" + e);
		}
	}

	/**
	 * 
	 * @param byteBuffer
	 * @param size
	 *            如果写入的字节数大于byteBuffer可写的部分则需要进行处理,要不然则直接返回即可。
	 * @param writeSocketIfFull
	 */
	public ByteBuffer checkWriteByteBuffer(ByteBuffer byteBuffer, int size, boolean writeSocketIfFull) {
		if (size > byteBuffer.remaining()) {
			if (writeSocketIfFull) {
				writeButNotSend(byteBuffer);
				return this.processor.getByteBufferPool().allocate(size);
			} else {// Relocate a larger buffer
				byteBuffer.flip();
				ByteBuffer newByteBuffer = this.processor.getByteBufferPool().allocate(size + byteBuffer.limit() + 1);
				newByteBuffer.put(byteBuffer);
				this.recycle(byteBuffer);
				return newByteBuffer;
			}
		} else {
			return byteBuffer;
		}
	}

	public ByteBuffer writeToByteBuffer(byte[] bytes, ByteBuffer byteBuffer) {
		int offset = 0;
		int length = bytes.length;
		int remaining = byteBuffer.remaining();
		while (length > 0) {
			if (remaining >= length) {
				byteBuffer.put(bytes, offset, length);
				break;
			} else {
				byteBuffer.put(bytes, offset, remaining);
				writeButNotSend(byteBuffer);
				byteBuffer = allocate();
				offset += remaining;
				length -= remaining;
				remaining = byteBuffer.remaining();
				continue;
			}
		}
		return byteBuffer;
	}

	@Override
	public void close(String reason) {
		if (!this.isClosed.get()) {
			closeSocketChannel();
			this.isClosed.set(true);
			if (this.processor != null) {
				this.processor.removeSessionContext(this);
			}
			this.cleanup();

			// ignore null information
			if (Strings.isNullOrEmpty(reason)) {
				return;
			}
			LOGGER.info("close connection,reason:" + reason + " ," + this);
			if (reason.contains("connection,reason:java.net.ConnectException")) {
				throw new RuntimeException(" errr");
			}
		}
	}

	public boolean isClosed() {
		return this.isClosed.get();
	}

	public void idleCheck() {
		if (isIdleTimeout()) {
			close(" idle ");
		}
	}

	/**
	 * 清理资源:那些是Session上下文的资源
	 */
	protected void cleanup() {

		// 清理资源占用
		if (this.readByteBuffer != null) {
			this.recycle(this.readByteBuffer);
			this.readByteBuffer = null;
			this.readByteBufferOffset = 0;
		}

		if (this.writeByteBuffer != null) {
			recycle(this.writeByteBuffer);
			this.writeByteBuffer = null;
		}

		if (!this.decompressUnfinishedDataQueue.isEmpty()) {
			this.decompressUnfinishedDataQueue.clear();
		}

		if (!this.compressUnfinishedDataQueue.isEmpty()) {
			this.compressUnfinishedDataQueue.clear();
		}

		ByteBuffer byteBuffer = null;
		while ((byteBuffer = this.writeByteBufferQueue.poll()) != null) {
			recycle(byteBuffer);
		}
	}

	/**
	 * 从指定的偏移量处读取包的长度
	 */
	protected int getPacketLength(ByteBuffer byteBuffer, int offset) {
		int headerSize = getPacketHeaderSize();
		headerSize = 7;
		/**
		 * 数据包错误
		 */
		if (byteBuffer.position() < offset + headerSize) {
			return -1;
		} else {
			int length = byteBuffer.get(offset) & 0xff;
			length |= (byteBuffer.get(++offset) & 0xff) << 8;
			length |= (byteBuffer.get(++offset) & 0xff) << 16;
			return length + headerSize;
		}
	}

	public ConcurrentLinkedQueue<ByteBuffer> getWriteByteBufferQueue() {
		return writeByteBufferQueue;
	}

	private void closeSocketChannel() {
		if (socketChannel != null) {

			boolean isSocketClosed = true;
			try {
				socketChannel.close();
			} catch (Exception e) {
				LOGGER.error("AbstractConnectionCloseError", e);
			}

			boolean closed = isSocketClosed && (!socketChannel.isOpen());
			if (closed == false) {
				LOGGER.error("close socket of connnection failed " + this);
			}
		}
	}

}
